package com.ihr360.binlog.daemon;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.Message;
import com.ihr360.binlog.dto.CanalConfig;
import com.ihr360.binlog.service.IProcessErrorCallback;
import com.ihr360.binlog.service.IProcessEventEntry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


public class BinlogReceiver {
    Logger logger = LoggerFactory.getLogger(BinlogReceiver.class.getName());
    private boolean isDeamon = false;

    private CanalConfig config;

    private IProcessEventEntry processEventEntry;

    private IProcessErrorCallback processErrorCallback;

    private ExecutorService executorService;

    public BinlogReceiver(CanalConfig config, IProcessEventEntry processEventEntry, IProcessErrorCallback errorCallback) {
        this.config = config;
        this.processEventEntry = processEventEntry;
        this.processErrorCallback = errorCallback;
    }

    int emptyCount = 0;

    public String start() {
        if (isDeamon) {
            return "It's Running Now";
        }
        isDeamon = true;
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        executorService.submit(() -> {
            // 创建链接
            CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(config.getHost(),
                    config.getPort()), config.getDestination(), config.getUsername(), config.getPassword());
            try {
                int batchSize = 1000;
                connector.connect();
                connector.subscribe(config.getFilter());
                connector.rollback();
                while (isDeamon) {
                    Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                    long batchId = message.getId();
                    int size = message.getEntries().size();
                    if (batchId == -1 || size == 0) {
                        emptyCount++;
                        if (logger.isInfoEnabled()) {
                            logger.info(config.getDestination()+ " empty count : " + emptyCount);
                        }
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            if (logger.isErrorEnabled()) {
                                logger.error(e.getMessage(), e.getCause());
                            }
                        } finally {
                            connector.ack(batchId); // 提交确认
                        }
                    } else {
                        emptyCount = 0;
                        try {
                            boolean success = processEventEntry.processEventEntrys(message.getEntries(), config);
                            if (success) {
                                connector.ack(batchId); // 提交确认
                            } else {
                                connector.rollback(batchId); // 处理失败, 回滚数据
                            }
                        } catch (Exception e) {
                            if (logger.isErrorEnabled()) {
                                logger.error("processEventEntrys exception", e.getCause());
                            }
                            connector.rollback(batchId);
                        }
                    }
                }
            } catch (Exception e) {
                isDeamon= false;
                processErrorCallback.onError(config,e);
                if (logger.isErrorEnabled()) {
                    logger.error(e.getMessage(), e.getCause());
                }
            } finally {
                connector.disconnect();
            }
        });
        return "Start Completed";
    }

    public void stop() {
        if (executorService.isShutdown()) {
            return;
        }
        isDeamon = false;
        executorService.shutdown();
    }

    public boolean isOn() {
        return isDeamon;
    }

    public String status() {
        return executorService.isShutdown() ? "DOWN" : "UP";
    }

}